/*
 *   Copyright 1999-2016 Asiainfo Technologies(China),Inc.
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 *
 */

package com.asiainfo.redis.service.busidatawriter;

import com.asiainfo.redis.entity.BusiDataRule;
import com.asiainfo.redis.service.BusiConfigService;
import com.asiainfo.redis.service.RedisService;
import com.asiainfo.redis.service.clusterhandler.RedisClusterHandler;
import com.asiainfo.redis.utils.SpringContextUtil;
import com.asiainfo.redis.web.PageData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.domain.Pageable;
import org.springframework.data.redis.connection.RedisClusterConnection;
import org.springframework.util.Assert;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @version v 1.0 on 2016/7/27 16:45
 * @auther william.xu
 */
//@Component
public class BusiDataHandler extends RedisClusterHandler {

    protected static final Logger LOGGER = LoggerFactory.getLogger(BusiDataHandler.class);

//    @Resource
    private BusiConfigService busiConfigService;

    private BusiDataRule busiDataRule;

    public BusiDataHandler(BusiDataRule busiDataRule) throws Exception{
        Assert.notNull(busiDataRule);
        this.busiDataRule = busiDataRule;
        busiConfigService = SpringContextUtil.getBean(BusiConfigService.class);
    }

    public void removeAllRedisData(Long clusterId) throws Exception{
        BusiDataRedisWriter busiDataRedisWriter = new BusiDataRedisWriter(0) {
            @Override
            public void doWrite(ResultSet rst) throws Exception {
            }
        };
        RedisClusterConnection redisClusterConnection = null;
        try {
            redisClusterConnection = getRedisClusterConnection(clusterId);
            busiDataRedisWriter.setConnection(redisClusterConnection);
            busiDataRedisWriter.setBusiDataRule(busiDataRule);
            busiDataRedisWriter.removeAllRedisDataByBusiTypeCodeAndKeyPrefix();
        }finally {
            if (redisClusterConnection != null) {
                closeRedisClusterConnection(redisClusterConnection);
            }
        }
    }

    public void queryDataAndWriteToRedis(Boolean isAllUpdate, String sql, Long clusterId, BusiDataRedisWriter busiDataRedisWriter)
            throws Exception{

        RedisClusterConnection redisClusterConnection = null;

        Connection connection = null;
        Statement statement = null;
        ResultSet rst = null;

        try {
            Long datasourceId = busiDataRule.getDatasourceId();
            connection = busiConfigService.getConnection(datasourceId);
            statement = connection.createStatement();
            //LOGGER.info("query sql=" + sql);
            rst = statement.executeQuery(sql);
            redisClusterConnection = getRedisClusterConnection(clusterId);
            busiDataRedisWriter.setConnection(redisClusterConnection);
            busiDataRedisWriter.setBusiDataRule(busiDataRule);
            if (isAllUpdate){
                //认为是全量刷新
                //spring jedis api调用keys方法是异步调用，跟多线程全量刷新会冲突，所以不在这里调用
                //busiDataRedisWriter.removeAllRedisDataByBusiTypeCodeAndKeyPrefix();
            }

            busiDataRedisWriter.doWrite(rst);

        }finally {
            if (connection != null){
                try {
                    connection.close();
                }catch (Exception e){
                }
            }
            if (statement != null) {
                try {
                    statement.close();
                }catch (Exception e){
                }
            }
            if (rst != null) {
                rst.close();
            }
            if (redisClusterConnection != null) {
                closeRedisClusterConnection(redisClusterConnection);
            }
        }

    }

    public PageData<Map> getDatabasePageData(String sql, Pageable pageable) throws Exception{

        Connection connection = null;
        Statement statement = null;
        ResultSet rst = null;
        PageData<Map> result = new PageData<>();

        try {
            Long datasourceId = busiDataRule.getDatasourceId();
            connection = busiConfigService.getConnection(datasourceId);
            statement = connection.createStatement();
            String pagesql = getPageSQL(sql, pageable.getPageNumber(), pageable.getPageSize());
            rst = statement.executeQuery(pagesql);
            List<Map> dataList = new ArrayList();
            String[] saveFields = busiDataRule.getSaveFields().split(BusiConfigService.SPLIT_CODE);
            while (rst.next()) {
                Map row = new HashMap();
                for (String saveField : saveFields) {
                    row.put(saveField, rst.getObject(saveField));
                }
                dataList.add(row);
            }
            result.setRows(dataList);
            //By default, only one ResultSet object per Statement object can be open at the same time.
            rst.close();
            rst = statement.executeQuery("select count(1) from ("+sql+")");
            if (rst.next()){
                String count = rst.getObject(1).toString();
                result.setTotal(Long.parseLong(count));
            }
            return result;
        }finally {
            if (connection != null){
                try {
                    connection.close();
                }catch (Exception e){
                }
            }
            if (statement != null) {
                try {
                    statement.close();
                }catch (Exception e){
                }
            }
            if (rst != null) {
                rst.close();
            }
        }

    }


    private String getPageSQL(String querySql, int pageNumber, int pageSize) {
        //这里pageNumber从0开始，pageSize是每页条数(根据org.springframework.data.domain.Pageable接口)
        int startNum = pageNumber * pageSize + 1;
        int endNum = startNum + pageSize - 1;

        // get pageList sql
        StringBuffer sb = new StringBuffer(
                "select tt.* from (select rownum num, t.* from (");
        sb.append(querySql);
        sb.append(") t where rownum<= ");
        sb.append(endNum);
        sb.append(") tt where num >= ");
        sb.append(startNum);
        String pageSql = sb.toString();

        return pageSql;
    }

    /**
     *
     * @param querySql
     * @return
     */
    private String getCountSql(String querySql) {
        int sqlFrom = querySql.indexOf(" FROM ");
        int sqlOrderby = querySql.lastIndexOf("ORDER BY");
        String countStr = "";

        // get count sql
        if (sqlOrderby > 0) {
            countStr = "select count(*) "
                    + querySql.substring(sqlFrom, sqlOrderby);
        } else {
            countStr = "select count(*) " + querySql.substring(sqlFrom);
        }

        return countStr;
    }

}
